Feat (Core): PostgresJobQueue Implementation and Job Processing Enhancements #30175

jgambarios · 2024-09-27T16:33:37Z

This PR introduces the PostgresJobQueue implementation, fulfilling the requirement for a default job queue using PostgreSQL.
In the process of developing this implementation, we've made significant improvements to the overall job processing system, including enhancements to the JobQueueManagerAPIImpl and related components.

Key Features of `PostgresJobQueue`:

Efficient Job Management: Implements all required operations for job queue management, including creating, retrieving, updating, and monitoring jobs.
JSONB Field Usage: Utilizes PostgreSQL's JSONB fields for storing job parameters and results, allowing for flexible and efficient data storage.
Optimized Queries: Implements efficient SQL queries, including the use of SELECT FOR UPDATE SKIP LOCKED for concurrent job processing.
Pagination Support: Provides paginated results for job listings, improving performance for large job sets.
Comprehensive Job Lifecycle Management: Handles all stages of a job's lifecycle, from creation to completion or failure.
Transactional Integrity: Ensures data consistency through proper transaction management.

Improvements to Job Processing System:

Enhanced Error Handling: Implemented a more robust error handling mechanism, including the use of custom exceptions for different error scenarios.
Improved Retry Logic: Refined the retry mechanism to handle transient failures more gracefully.
Real-Time Job Monitoring: Enhanced the real-time monitoring capabilities, allowing for more responsive job status updates.
Progress Tracking: Improved the progress tracking system, providing more accurate and timely updates on job completion percentage.
Cancellation Support: Implemented a more reliable job cancellation process.

`JobQueueManagerAPIImpl` Enhancements:

Event-Driven Updates: Implemented an event-driven system for job status updates, improving system responsiveness.
More Granular Job State Transitions: Implemented more detailed job state transitions, providing better insights into job processing stages.

Testing and Documentation:

Developed an extensive set of unit and integration tests for the PostgresJobQueue implementation.

Simple class diagram with main operations and interactions

classDiagram
    class JobQueueManagerAPI {
        <<interface>>
        +start()
        +close()
        +createJob(queueName: String, parameters: Map): String
        +getJob(jobId: String): Job
        +getJobs(page: int, pageSize: int): JobPaginatedResult
        +cancelJob(jobId: String)
        +watchJob(jobId: String, watcher: Consumer~Job~)
        +setRetryStrategy(queueName: String, retryStrategy: RetryStrategy)
    }

    class JobQueueManagerAPIImpl {
        -jobQueue: JobQueue
        -processors: Map~String, JobProcessor~
        -retryStrategies: Map~String, RetryStrategy~
        -circuitBreaker: CircuitBreaker
        +start()
        +close()
        -processJobs()
        -processJobWithRetry(job: Job)
    }

    class JobQueue {
        <<interface>>
        +createJob(queueName: String, parameters: Map): String
        +getJob(jobId: String): Job
        +getActiveJobs(queueName: String, page: int, pageSize: int): JobPaginatedResult
        +getCompletedJobs(queueName: String, startDate: LocalDateTime, endDate: LocalDateTime, page: int, pageSize: int): JobPaginatedResult
        +updateJobStatus(job: Job)
        +nextJob(): Job
        +updateJobProgress(jobId: String, progress: float)
    }

    class PostgresJobQueue {
        -objectMapper: ObjectMapper
        +createJob(queueName: String, parameters: Map): String
        +getJob(jobId: String): Job
        +getActiveJobs(queueName: String, page: int, pageSize: int): JobPaginatedResult
        +getCompletedJobs(queueName: String, startDate: LocalDateTime, endDate: LocalDateTime, page: int, pageSize: int): JobPaginatedResult
        +updateJobStatus(job: Job)
        +nextJob(): Job
        +updateJobProgress(jobId: String, progress: float)
    }

    class Job {
        +id: String
        +queueName: String
        +state: JobState
        +parameters: Map
        +progress: float
        +createdAt: LocalDateTime
        +updatedAt: LocalDateTime
        +startedAt: Optional~LocalDateTime~
        +completedAt: Optional~LocalDateTime~
        +result: Optional~JobResult~
    }

    class JobProcessor {
        <<interface>>
        +process(job: Job)
        +getResultMetadata(job: Job): Map
    }

    class Cancellable {
        <<interface>>
        +cancel(job: Job)
    }

    class RetryStrategy {
        <<interface>>
        +shouldRetry(job: Job, exceptionClass: Class): boolean
        +nextRetryDelay(job: Job): long
    }

    class CircuitBreaker {
        -failureThreshold: int
        -resetTimeout: long
        +allowRequest(): boolean
        +recordFailure()
        +reset()
    }

    class RealTimeJobMonitor {
        -jobWatchers: Map~String, List~Consumer~Job~~~
        +registerWatcher(jobId: String, watcher: Consumer~Job~)
        +updateWatchers(updatedJobs: List~Job~)
    }

    JobQueueManagerAPI <|.. JobQueueManagerAPIImpl
    JobQueueManagerAPIImpl --> JobQueue
    JobQueueManagerAPIImpl --> JobProcessor
    JobQueueManagerAPIImpl --> RetryStrategy
    JobQueueManagerAPIImpl --> CircuitBreaker
    JobQueueManagerAPIImpl --> RealTimeJobMonitor
    JobQueue <|.. PostgresJobQueue
    JobQueueManagerAPIImpl ..> Job
    JobProcessor <|-- Cancellable

This PR fixes: #29479

Introduced multiple job lifecycle event classes including `JobCreatedEvent`, `JobStartedEvent`, `JobProgressUpdatedEvent`, `JobCompletedEvent`, `JobFailedEvent`, and `JobCancelledEvent`. Updated `JobQueueManagerAPIImpl` to fire these events at appropriate stages in the job lifecycle to enhance observability and event-driven processing. Updated tests to accommodate the changes in event handling.

Replaced jobId strings with Job objects in event classes to simplify data access and reduce redundancy. Introduced RealTimeJobMonitor for handling real-time job updates, improving event notification efficiency and code maintainability. Updated tests and related classes to accommodate these changes.

Remove the `JobResult` enum and update job processing methods to use an `AbstractJobResult` interface. This refactor centralizes job result handling, differentiates success, failure, and cancellation cases, and incorporates metadata and error details within the job result.